/**
 * Copyright 2005 The Apache Software Foundation
 * <p>
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.hadoop.mapred;

import org.apache.hadoop.io.*;
import org.apache.hadoop.ipc.*;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.util.*;

import java.io.*;
import java.net.*;
import java.util.*;
import java.util.logging.*;

/**
 * Runs a reduce task.
 */
class ReduceTaskRunner extends TaskRunner {
    private static final Logger LOG =
            LogFormatter.getLogger("org.apache.hadoop.mapred.ReduceTaskRunner");
    private final MapOutputFile mapOutputFile;

    public ReduceTaskRunner(Task task, TaskTracker tracker, Configuration conf) {
        super(task, tracker, conf);
        this.mapOutputFile = new MapOutputFile();
        this.mapOutputFile.setConf(conf);
    }

    /**
     * Assemble all of the map output files.
     */
    @Override
    public boolean prepare() throws IOException {
        ReduceTask task = ((ReduceTask) getTask());
        // cleanup from failures
        this.mapOutputFile.removeAll(task.getTaskId());
        String[][] mapTaskIds = task.getMapTaskIds();
        final Progress copyPhase = getTask().getProgress().phase();

        // we need input from every map task
        Vector needed = new Vector();
        for (String[] mapTaskId : mapTaskIds) {
            needed.add(mapTaskId);
            // add sub-phase per file
            copyPhase.addPhase();
        }

        InterTrackerProtocol jobClient = getTracker().getJobClient();
        while (needed.size() > 0) {
            getTask().reportProgress(getTracker());

            // query for a just a random subset of needed segments so that we don't
            // overwhelm jobtracker.  ideally perhaps we could send a more compact
            // representation of all needed, i.e., a bit-vector
            Collections.shuffle(needed);
            int checkSize = Math.min(10, needed.size());
            String[][] neededStrings = new String[checkSize][];
            for (int i = 0; i < checkSize; i++) {
                neededStrings[i] = (String[]) needed.elementAt(i);
            }
            MapOutputLocation[] locs =
                    jobClient.locateMapOutputs(task.getTaskId(), neededStrings);

            if (locs.length == 0) {
                try {
                    if (killed) {
                        return false;
                    }
                    Thread.sleep(10000);
                } catch (InterruptedException ignored) {
                }
                continue;
            }

            LOG.info(task.getTaskId() + " Got " + locs.length + " map output locations.");

            // try each of these locations
            for (MapOutputLocation loc : locs) {
                InetSocketAddress addr =
                        new InetSocketAddress(loc.getHost(), loc.getPort());
                MapOutputProtocol client =
                        (MapOutputProtocol) RPC.getProxy(MapOutputProtocol.class, addr, this.conf);

                this.mapOutputFile.setProgressReporter(progress -> {
                    copyPhase.phase().set(progress);
                    getTask().reportProgress(getTracker());
                });

                getTask().reportProgress(getTracker());
                try {
                    copyPhase.phase().setStatus(loc.toString());

                    client.getFile(loc.getMapTaskId(), task.getTaskId(), new IntWritable(task.getPartition()));

                    // Success: remove from 'needed'
                    boolean foundit = false;
                    for (Iterator it = needed.iterator(); it.hasNext() && !foundit; ) {
                        String[] idsForSingleMap = (String[]) it.next();
                        for (String s : idsForSingleMap) {
                            if (s.equals(loc.getMapTaskId())) {
                                it.remove();
                                foundit = true;
                                break;
                            }
                        }
                    }
                    copyPhase.startNextPhase();

                    // failed: try again later
                } finally {
                    this.mapOutputFile.setProgressReporter(null);
                }
            }
        }
        getTask().reportProgress(getTracker());
        return true;
    }

    /**
     * Delete all of the temporary map output files.
     */
    @Override
    public void close() throws IOException {
        getTask().getProgress().setStatus("closed");
        this.mapOutputFile.removeAll(getTask().getTaskId());
    }

}
